#pragma once

#include "../common/Common.h"
#include "../common/Base.h"
#include "../common/Message.h"
#include "Request.h"

namespace rpc
{
    namespace client
    {

        class TopicManager
        {
        public:
            using SubCallback = std::function<void(const string &, const string &)>;
            using ptr = std::shared_ptr<TopicManager>;

            TopicManager(const Request::ptr &req) : _request(req) {}

            bool create(const BaseConnection::ptr &conn, const string &topic)
            {
                DLOG("TopicManager create()");
                return commonRequest(conn, topic, TopicOpType::TOPIC_CREATE);
            }

            bool remove(const BaseConnection::ptr &conn, const string &topic)
            {
                return commonRequest(conn, topic, TopicOpType::TOPIC_DELETE);
            }

            bool subscribe(const BaseConnection::ptr &conn, const string &topic, const SubCallback &cb)
            {
                addSubscribe(topic, cb);
                if (!commonRequest(conn, topic, TopicOpType::TOPIC_SUBCRIBE))
                {

                    delSubscribe(topic);
                    return false;
                }

                return true;
            }

            bool cancel(const BaseConnection::ptr &conn, const string &topic)
            {
                delSubscribe(topic);
                return commonRequest(conn, topic, TopicOpType::TOPIC_CANCEL);
            }

            bool publish(const BaseConnection::ptr &conn, const string &topic, const string &msg)
            {
                return commonRequest(conn, topic, TopicOpType::TOPIC_PUBLISH, msg);
            }

            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &req)
            {
                auto type = req->opType();
                if (type != TopicOpType::TOPIC_PUBLISH)
                {
                    ELOG("收到错误的主题操作");
                    return;
                }

                string topic = req->topic();
                string msg = req->msg();
                auto cb = getCallback(topic);
                if (!cb)
                {
                    ELOG("收到了 %s 主题消息，但是该消息无主题处理回调！", topic.c_str());
                    return;
                }

                return cb(topic, msg);
            }

        private:
            void addSubscribe(const string &topic, const SubCallback &cb)
            {
                unique_lock<mutex> lock(_mtx);
                _topicCbs.emplace(topic, cb);
            }

            void delSubscribe(const string &topic)
            {
                unique_lock<mutex> lock(_mtx);
                _topicCbs.erase(topic);
            }

            const SubCallback getCallback(const string &topic)
            {
                unique_lock<mutex> lock(_mtx);
                auto it = _topicCbs.find(topic);
                if (it == _topicCbs.end())
                    return SubCallback();

                return it->second;
            }

            bool commonRequest(const BaseConnection::ptr &conn, const string &topic, TopicOpType type, const string msg = "")
            {
                // 构造请求对象并填充数据
                auto req = MessageFactory::create<TopicRequest>();
                req->setRid(uuid());
                req->setMtype(MType::REQ_TOPIC);
                req->setOpType(type);
                req->setTopic(topic);

                if (type == TopicOpType::TOPIC_PUBLISH)
                {
                    req->setMsg(msg);
                }

                // 向服务端发送，等待响应
                BaseMessage::ptr resp;
                if (!_request->send(conn, req, resp))
                {
                    ELOG("主题操作请求失败！");
                    return false;
                }

                DLOG("RpcTopic.h commonRequest() after send() ...");

                auto topicResp = std::dynamic_pointer_cast<TopicResponse>(resp);
                if (!topicResp)
                {
                    ELOG("主题操作响应，乡下类型转换失败！");
                    return false;
                }

                if (topicResp->rcode() != RCode::RCODE_OK)
                {
                    ELOG("主题操作请求出错 %s", errReason(topicResp->rcode()).c_str());
                    return false;
                }

                return true;
            }

        private:
            mutex _mtx;
            unordered_map<string, SubCallback> _topicCbs;
            Request::ptr _request;
        };

    };
};